package com.sunday.common.mq.rocket.brave.study.base.e12_POP;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageRequestMode;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.junit.jupiter.api.Test;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * https://baijiahao.baidu.com/s?id=1716353444200487221&wfr=spider&for=pc
 */
@Slf4j
public class PushPopConsumer {


    String consumerGroup = "push_pop_consumer_group_name";
    String namesrvAddr = "127.0.0.1:9876";
    String topic = "PushPopTopic";

    public Map<Integer, PushPopConsumerThread> popMap = new ConcurrentHashMap<>();

    /**
     * pull模式未实现pop
     */
    @Test
    public void pullConsumerReadQueue() throws Exception {
        for (int i = 0; i < 10; i++) {
            PullConsumerThread pullConsumerThread = new PullConsumerThread(namesrvAddr, consumerGroup, topic, i + "");
            pullConsumerThread.start();
        }

        TimeUnit.SECONDS.sleep(600);
    }

    @Test
    public void popConsumerReadQueue() throws Exception {
        for (int i = 0; i < 10; i++) {
            PushPopConsumerThread pushPopConsumerThread = new PushPopConsumerThread(consumerGroup, namesrvAddr, topic, i + "");
            pushPopConsumerThread.start();
            popMap.put(i, pushPopConsumerThread);
        }

        /**
         * 输出心跳
         */
        new Thread(() -> {
            while (true) {
                popMap.forEach((integer, pushPopConsumerThread) -> {
                    log.info("{} -> {}", pushPopConsumerThread.getNode(), pushPopConsumerThread.getCount());
                    pushPopConsumerThread.getCount().set(0);
                });
                try {
                    TimeUnit.SECONDS.sleep(10);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }).start();

        TimeUnit.SECONDS.sleep(600);
    }


    /**
     *
     * 使用命令行方式切换
     * 利用命令行，用如下命令，指定集群和需要切换的消费组，可以将一个消费组切换成 Pop 消费模式消费某个 Topic
     *
     * mqadmin setConsumeMode -c cluster -t topic -g group -m POP -q 8
     *
     * POP Client 从 Broker 的队列中发出 POP 请求消息，Broker 返回消息 message。
     * 在消息的系统属性里面有一个比较重要的属性叫做 POP_CK，POP_CK 为一条消息的 handler，通过一个 handler 就可以定位到一条消息。
     * 当消息消费成功之后，POP client 发送 ackMessage 并传递 handler 向 broker 确认消息消费成功。
     * <p>
     * Cluster -> config -> defaultMessageRequestMode = PULL 管理界面看不了具体参数
     * 具体修改完的参数值请使用命令行查看
     * MessageExt [brokerName=broker-single, queueId=2, storeSize=241, queueOffset=2570, sysFlag=0, bornTimestamp=1688712674342, bornHost=/172.18.0.1:57532, storeTimestamp=1688712674344, storeHost=/172.26.224.1:10911, msgId=AC1AE00100002A9F0000000000317025, commitLogOffset=3239973, bodyCRC=935238117, reconsumeTimes=0, preparedTransactionOffset=0, toString()=
     * Message{topic='PushPopTopic', flag=0,
     * properties={
     * CONSUME_START_TIME=1688712674346,
     * MSG_REGION=DefaultRegion,
     * 1ST_POP_TIME=1688712674345,
     * UNIQ_KEY=C0A88940671436BAF30C22161C26025B,
     * CLUSTER=DefaultCluster,
     * POP_CK=2570 1688712674345 60000 0 0 broker-single 2 2570, TAGS=TagA, WAIT=true, TRACE_ON=true
     * }, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32], transactionId='null'}]
     */
    @Test
    public void switchPop() {
        try {
            DefaultMQAdminExt admin = new DefaultMQAdminExt();
            // 设置NameServer地址
            admin.setNamesrvAddr(namesrvAddr);
            // 启动DefaultMQAdminExt
            admin.start();

            ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
            Set<String> brokerAddrs = clusterInfo.getBrokerAddrTable().values().stream().map(BrokerData::selectBrokerAddr).collect(Collectors.toSet());
            for (String brokerAddr : brokerAddrs) {
                log.info("{}", brokerAddr);
                // popShareQueueNum 在弹出模式下工作的消费者可以共享分配给cid列表中紧随其后的N (N = popShareQueueNum)消费者的MessageQueues
                admin.setMessageRequestMode(brokerAddr, topic, consumerGroup, MessageRequestMode.POP, 8, 3_000);
//                admin.setMessageRequestMode(brokerAddr, topic, consumerGroup, MessageRequestMode.PULL, 8, 3_000);
                // 订阅某个组的配置
                SubscriptionGroupConfig examineSubscriptionGroupConfig = admin.examineSubscriptionGroupConfig(brokerAddr, consumerGroup);
                log.info("{}", examineSubscriptionGroupConfig);

//                ConsumerConnection consumerConnection = admin.examineConsumerConnectionInfo(consumerGroup);
//                log.info("{}", consumerConnection);
//                log.info("{}", consumerConnection.getMessageModel());

                //获取订阅消费组列表
//                SubscriptionGroupWrapper subscriptionGroupWrapper = admin.getAllSubscriptionGroup(brokerAddr, 1000);
//                subscriptionGroupWrapper.getSubscriptionGroupTable().forEach((k, v) -> log.info("{} = {}", k, v));
            }

        } catch (Exception e) {
            log.error(e.toString(), e);
        }

    }

    /**
     * MessageExt
     * [
     * brokerName=broker-single, queueId=1, storeSize=241, queueOffset=2418, sysFlag=0, bornTimestamp=1688712046285,
     * bornHost=/172.18.0.1:53976, storeTimestamp=1688712046288, storeHost=/172.26.224.1:10911,
     * msgId=AC1AE00100002A9F00000000002A661D, commitLogOffset=2778653, bodyCRC=935238117, reconsumeTimes=0,
     * preparedTransactionOffset=0,
     * <p>
     * toString()=
     * Message{
     * topic='PushPopTopic', flag=0, properties={
     * CONSUME_START_TIME=1688712046290,
     * MSG_REGION=DefaultRegion,
     * UNIQ_KEY=C0A88940954036BAF30C220C86CD00E6,
     * CLUSTER=DefaultCluster,
     * MIN_OFFSET=0,
     * TAGS=TagA,
     * WAIT=true,
     * TRACE_ON=true,
     * MAX_OFFSET=2419
     * },
     * body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32],
     * transactionId='null'
     * }
     * ]
     */
    @Test
    public void switchPull() {
        try {
            DefaultMQAdminExt admin = new DefaultMQAdminExt();
            // 设置NameServer地址
            admin.setNamesrvAddr(namesrvAddr);
            // 启动DefaultMQAdminExt
            admin.start();

            ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
            Set<String> brokerAddrs = clusterInfo.getBrokerAddrTable().values().stream().map(BrokerData::selectBrokerAddr).collect(Collectors.toSet());
            for (String brokerAddr : brokerAddrs) {
                log.info("{}", brokerAddr);
                // popShareQueueNum 在弹出模式下工作的消费者可以共享分配给cid列表中紧随其后的N (N = popShareQueueNum)消费者的MessageQueues
                admin.setMessageRequestMode(brokerAddr, topic, consumerGroup, MessageRequestMode.PULL, 8, 3_000);
            }

        } catch (Exception e) {
            log.error(e.toString(), e);
        }

    }


}
